package study.dq.jms.p2p;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author dongqun
 * @version 1.0
 * @description 类的功能描述
 * @date 2019/10/18 18:15
 */
public class Customer {
    private QueueConnection connection;
    private QueueSession session;
    private ThreadLocal<MessageConsumer> threadLocal = new ThreadLocal<>();
    private AtomicInteger count = new AtomicInteger(0);
    public Customer(String username, String password, String borkerUrl){
        try {
            ConnectionFactory factory = new ActiveMQConnectionFactory(username,password,borkerUrl);
            connection = ((ActiveMQConnectionFactory) factory).createQueueConnection();
            connection.start();
            session = connection.createQueueSession(false,Session.CLIENT_ACKNOWLEDGE);
//            session.setMessageListener(this);
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    public void getMessage(String disname){
        try {
            Queue queue = session.createQueue(disname);
            QueueReceiver receiver;
            if (threadLocal.get()!=null){
                receiver = (QueueReceiver) threadLocal.get();
            }else{
                receiver = session.createReceiver(queue);
                threadLocal.set(receiver);
            }
//            while (true){
                Thread.sleep(1000);
                TextMessage textMessage = (TextMessage) receiver.receive();
                if (textMessage!=null){
                    System.out.println(Thread.currentThread().getName()+"我是消费者，我现在正在消费消息:"+textMessage.getText()+"count:"+count.getAndIncrement());
                    textMessage.acknowledge();
                }else {
                    System.out.println("---------------------------------------消费队列："+Thread.currentThread().getName()+"退出!");
//                    connection.close();
                }

//            }
        } catch (JMSException | InterruptedException e) {
            e.printStackTrace();
        }

    }

//    @Override
//    public void onMessage(Message message) {
//        TextMessage textMessage = (TextMessage) message;
//        try {
//            if (textMessage!=null){
//                System.out.println(Thread.currentThread().getName()+"我是消费者，我现在正在消费消息:"+textMessage.getText()+"count:"+count.getAndIncrement());
//            }else {
//                System.out.println("---------------------------------------消费队列："+Thread.currentThread().getName()+"退出!");
//            }
//        } catch (JMSException e) {
//            e.printStackTrace();
//        }
//    }

    public static void main(String[] args) {
        Customer customer = new Customer("admin","admin","tcp://127.0.0.1:61616");

    }
}
